package com.boarsoft.message.bean;

import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;

import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.message.core.MessageContext;
import com.boarsoft.message.core.MessageService;

/**
 * 实现TargetQueue，添加消息时会持久化（如果有Roller的话）
 * 
 * @author Mac_J
 *
 */
public class RecipientMQ extends MessageQueue implements Runnable {
	private static final Logger log = LoggerFactory.getLogger(RecipientMQ.class);

	/** 如果是配置的，会自动注入，反之是通过构造函数传入的 */
	@Autowired
	@Lazy
	protected MessageContext context;

	/** 与RPC的服务引用版本对应 */
	protected String version = "1.0.0";
	/** */
	protected int maxRetry = 3;
	/** 重试次数 */
	protected int retry = 3;
	/** 重试间隔 */
	protected int interval = 1000;
	/** 是否动态创建的，动态创建的收件人队列需要在没服务提供者是动态删除 */
	protected boolean dynamic;
	/** 消费与发外线程数 */
	protected int threads = 10;

	protected String sender;

	protected volatile boolean running = true;

	public RecipientMQ() {
	}

	public RecipientMQ(String code) {
		this.code = code;
	}

	@Override
	public String toString() {
		return new StringBuilder().append(code).append(":").append(catalog)//
				.append("/").append(name).append("/").append(version).toString();
	}

	public void handle(MessageEntity me) throws InterruptedException {
		// if (queue.add(me)) {
		// return;
		// }
		// log.warn("Can not add message {} to {} because its full", me, code);
		queue.put(me);
	}

	/**
	 * 如果是点对点消息，直接远程调用消息接收者的put法，将消息写入其消费队列<br>
	 * 如果是TOPIC消息，远程调用message-server的write方法，将消息写入发送队列
	 */
	@Override
	public void run() {
		running = true;
		MessageEntity me = null;
		while (running) {
//			if (queue == null) {
//				log.warn("Sleep 1 second since queue is still null");
//				try {
//					Thread.sleep(1000L);
//				} catch (InterruptedException e) {
//					// Nothing to do
//				}
//				continue;
//			}
			try {
				me = queue.poll(1L, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				log.error("Error on take message from recipient queue", e);
				continue;
			}
			if (me == null) {
				continue;
			}
			try {
				this.send(me);
			} catch (Throwable e) {
				log.error("Error on send message {}", me, e);
			}
		}
	}

	public void send(MessageEntity me) {
		Message m = me.getMessage();
		// 判断此消息的各收件人是否已成功发送，成功是不再发送
		RecipientStatus rs = m.getRecipientStatus(code);
		if (rs.getStatus() == RecipientStatus.SUCCESS) {
			return;
		}
		if (rs.getTried() >= maxRetry) {
			// 下次重启时会再重发，如果不希望再重发，则人工删除之
			return;
		}
		// 开始发送
		while (true) {
			int t = m.tried1(code);
			log.debug("Send message {} to {} at {}/{}", m, sender, t, maxRetry);
			MessageService ms = context.getConsumer(sender);
			if (ms == null) {
				log.error("No message service {} for message {}, retry in {}ms", //
						sender, m, interval);
			} else {
				try {
					ms.put(m);
					this.onRecipientEvent(me, RecipientStatus.SUCCESS, t, "");
					return;
				} catch (Exception e) {
					if (maxRetry < 0 || t < maxRetry) {
						log.error("Send message {} to {} failed, retry in {}ms", //
								m, code, interval, e);
					} else {
						log.error("Send message {} to {} failed, give up", m, code, e);
					}
					// 暂时将发送失败的消息，置为PENDING，以便重启时重发，如果不希望重发，则人工删除之
					this.onRecipientEvent(me, RecipientStatus.PENDING, t, e.getMessage());
				}
			}
			if (maxRetry < 0 || t < maxRetry) {
				if (interval > 0) {
					try {
						Thread.sleep(interval);
					} catch (InterruptedException e1) {
						log.error("Be interrupted while wait to send message", e1);
					}
				}
				continue;
			}
			return;
		}
	}

	/**
	 * 消息事件处理，目前仅当某一条消息从本地外发失败时，会调用此方法<br>
	 * 注：当通过服务器签收（转发失败的通知）或者收件人之一签收时<br>
	 * 显示调用的是message的acknowledge属性所配置的MessageService服务<br>
	 * 
	 * @param me
	 *            需要签收的原始消息
	 * @param status
	 *            （相对于该收件人）该消息是否送达
	 * @param tried
	 *            尝试次数
	 * @param memo
	 *            原因或其它备注
	 */
	public void onRecipientEvent(MessageEntity me, short status, int tried, String memo) {
		Message m = me.getMessage();
		// 更新消息的收件状态，如果所有收件人状态都完成态，则更新消息状态
		try {
			me.setRecipientStatus(code, status, tried, memo);
		} catch (Exception e) {
			log.error("Error on update message entity {}", m, e);
		}
		// 如果消息不需要签收，则返回
		if (!m.isNeedAcknowledge()) {
			return;
		}

		// 准备发送签收消息
		String target = m.getAcknowledge();
		Message a = new Message();
		a.setId(RandomUtil.genUUID());
		a.setAttachment(m);
		a.setKey(code); // recipient
		a.setContent(memo);
		a.addRecipient(m.getFrom(), true);
		a.setTarget(target);

		// 否则（作为消息服务器）转发此签收消息到原始的发件人
		String from = m.getFrom();
		log.info("Transmit message acknowledge {} to {}", a, from);
		// 获取当时用于发送此消息的targetQueue，通过它和它的roller更新消息状态
		try {
			m.setStatus(Message.STATUS_ACK_SENDING);
			context.getConsumer(from).put(a);
			m.setStatus(Message.STATUS_ACK_SENT);
		} catch (Exception e) {
			log.error("Error on acknowledge to {}", from, m.getId(), e);
			memo = e.getMessage();
		}
		try {
			me.setRecipientStatus(code, Message.STATUS_ACK_FAILED, m.getTried(), memo);
		} catch (Exception e) {
			log.error("Error on update message entity {}", m, e);
		}
	}

	public RecipientQueueInfo getQueueInfo() {
		RecipientQueueInfo r = new RecipientQueueInfo();
		r.setCapacity(capacity);
		r.setCatalog(catalog);
		r.setCode(code);
		r.setName(name);
		r.setPriority(priority);
		if (roller != null) {
			r.setRoller(roller.toString());
		}
		//
		r.setRetry(retry);
		r.setInterval(interval);
		return r;
	}

	// -------------

	public int getRetry() {
		return retry;
	}

	public void setRetry(int retry) {
		if (retry > 3) {
			throw new IllegalArgumentException("Retry must <= 3");
		}
		this.retry = retry;
	}

	public int getInterval() {
		return interval;
	}

	public void setInterval(int interval) {
		this.interval = interval;
	}

	public int getMaxRetry() {
		return maxRetry;
	}

	public void setMaxRetry(int maxRetry) {
		this.maxRetry = maxRetry;
	}

	public MessageContext getContext() {
		return context;
	}

	public void setContext(MessageContext context) {
		this.context = context;
	}

	public String getVersion() {
		return version;
	}

	public void setVersion(String version) {
		this.version = version;
	}

	public boolean isRunning() {
		return running;
	}

	public void stop() {
		this.running = false;
	}

	public boolean isDynamic() {
		return dynamic;
	}

	public void setDynamic(boolean dynamic) {
		this.dynamic = dynamic;
	}

	public String getSender() {
		return sender;
	}

	public void setSender(String sender) {
		this.sender = sender;
	}

	public int getThreads() {
		return threads;
	}

	public void setThreads(int threads) {
		this.threads = threads;
	}

}